package studio.raptor.databus.client;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.cmdb.Config;
import studio.raptor.cmdb.ConfigService;

import javax.jms.JMSException;
import javax.jms.Session;

/**
 * 队列消息生产者，每个生产者只能发送消息到一个指定的主题
 * <p>
 *     默认配置如下:<br>
 *     发送时间默认不超时<br>
 *     failover尝试重连次数一直重连<br>
 *     failover两次重连之间最大时间为10s<br>
 *     开启事务性会话<br>
 *     使用自动ACK模式<br>
 * </p>
 */
public class DatabusProducerBuilder {
  private static Logger log = LoggerFactory.getLogger(DatabusProducerBuilder.class);

    /**
     * 发送操作的超时时间，此时间并不会因重连操作打断。-1表示不超时。
     */
    private int timeout = -1;
    /**
     * 两次重连之间最长时间间隔，10s.
     */
    private int maxReconnectDelay = 10000;

    /**
     * failover机制中的最大重连次数。-1表示一直重连。
     */
    private int maxReconnectAttempts = -1;
    /**
     * 客户端启动时最大重连次数，默认为5
     */
    private int startupMaxReconnectAttempts = 5;

    private String otherURLConfig = "&";

    /**
     * 会话是否为事务性，生产者默认开启
     */
    private boolean transactedSession = true;

    /**
     * 消息ACK模式，默认值(1)为自动确认
     */
    private int acknowledgeMode = Session.AUTO_ACKNOWLEDGE ;


  /**
   * 发送消息的有效期，单位天
   */
  private int msgTimeToLive = 7 ;


    private DatabusConfig databusConfig ;

    public DatabusProducerBuilder(DatabusConfig databusConfig){
        this.databusConfig = databusConfig;
    }



    /**
     * 创建生产者
     * @return
     */
    public DatabusProducer create() throws JMSException {
        DatabusProducer producer
            = new DatabusProducer(databusConfig,this,this.databusConfig.getBrokerGroups());
        //初始化生产者配置
      try {
        initConfig();
      } catch (IOException e) {
        log.error(e.getMessage());
        throw new JMSException(e.getMessage());
      }
      return producer ;
    }

    /**
     * 依据配置加载方式初始化生产者配置
     */
    private void initConfig() throws IOException {
        this.setAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
        if( this.databusConfig.getLoadType() == DatabusConfigLoadType.LOAD_TYPE_CMDB ){
            Config config = ConfigService.getConfig(DatabusConfig.CLIENT_CONFIG_PRE+databusConfig.getBizKey());
            this.setStartupMaxReconnectAttempts(config.getIntProperty("producer.startupMaxReconnectAttempts",this.getStartupMaxReconnectAttempts()));
            this.setMaxReconnectAttempts(config.getIntProperty("producer.maxReconnectAttempts",this.getMaxReconnectAttempts()));
            this.setOtherURLConfig(config.getProperty("producer.otherURLConfig",this.getOtherURLConfig()));
            this.setTimeout(config.getIntProperty("producer.timeout",this.getTimeout()));
            this.setTransactedSession(config.getBooleanProperty("producer.transactedSession",this.isTransactedSession()));
            this.setMaxReconnectDelay(config.getIntProperty("producer.maxReconnectDelay",this.getMaxReconnectDelay()));
            this.setMsgTimeToLive(config.getIntProperty("producer.msgTimeToLive",this.getMsgTimeToLive()));
        }else{

            InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream(this.databusConfig.CONFIG_LOCAL_DIR + "producer.properties");
            Properties properties = new Properties();
            properties.load(in);
            this.setStartupMaxReconnectAttempts(Integer.valueOf( properties.getProperty("producer.startupMaxReconnectAttempts")));
            this.setMaxReconnectAttempts(Integer.valueOf( properties.getProperty("producer.maxReconnectAttempts")));
            this.setOtherURLConfig(properties.getProperty("producer.otherURLConfig"));
            this.setTimeout(Integer.valueOf( properties.getProperty("producer.timeout")));
            this.setTransactedSession("true".equals(properties.getProperty("producer.transactedSession")));
            this.setMaxReconnectDelay(Integer.valueOf( properties.getProperty("producer.maxReconnectDelay")));
            if( null != properties.getProperty("producer.msgTimeToLive") ){
              this.setMsgTimeToLive(Integer.valueOf(properties.getProperty("producer.msgTimeToLive")));
            }
        }

    }


     int getTimeout() {
        return timeout;
    }

     void setTimeout(int timeout) {
        this.timeout = timeout;
    }

     int getMaxReconnectDelay() {
        return maxReconnectDelay;
    }

     void setMaxReconnectDelay(int maxReconnectDelay) {
        this.maxReconnectDelay = maxReconnectDelay;
    }

     int getMaxReconnectAttempts() {
        return maxReconnectAttempts;
    }

     void setMaxReconnectAttempts(int maxReconnectAttempts) {
        this.maxReconnectAttempts = maxReconnectAttempts;
    }

     String getOtherURLConfig() {
        return otherURLConfig;
    }

     void setOtherURLConfig(String otherURLConfig) {
        this.otherURLConfig = otherURLConfig;
    }

     boolean isTransactedSession() {
        return transactedSession;
    }

     void setTransactedSession(boolean transactedSession) {
        this.transactedSession = transactedSession;
    }

     int getAcknowledgeMode() {
        return acknowledgeMode;
    }

     void setAcknowledgeMode(int acknowledgeMode) {
        this.acknowledgeMode = acknowledgeMode;
    }

     int getStartupMaxReconnectAttempts() {
        return startupMaxReconnectAttempts;
    }

  /**
   * 获取消息保留天数
   * @return
   */
  int getMsgTimeToLive() {
    return msgTimeToLive;
  }

  void setMsgTimeToLive(int msgTimeToLive) {
    this.msgTimeToLive = msgTimeToLive;
  }

  /**
     * 设置客户端启动时尝试连接服务器的最大次数，默认为5
     * @param startupMaxReconnectAttempts 客户端启动时尝试连接服务器的最大次数
     */
    public void setStartupMaxReconnectAttempts(int startupMaxReconnectAttempts) {
        this.startupMaxReconnectAttempts = startupMaxReconnectAttempts;
    }

    /**
     * 将生产者客户端配置转换为URL参数
     * @return
     */
    public String transform2URLConfig(){
        StringBuilder brokerURLConfig = new StringBuilder();
        brokerURLConfig.append("maxReconnectDelay=" + this.getMaxReconnectDelay());
        brokerURLConfig.append("&maxReconnectAttempts=" + this.getMaxReconnectAttempts());
        brokerURLConfig.append(this.getOtherURLConfig());
        return brokerURLConfig.toString();
    }
}
